package com.qyb.mq.producer;

import com.qyb.util.JsonUtils;
import com.fasterxml.jackson.core.JsonProcessingException;
import com.fasterxml.jackson.databind.ObjectMapper;
import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.ProducerRecord;

import java.util.Map;
import java.util.Properties;
import java.util.concurrent.ExecutionException;

/**
 * *************************************************************************************************
 * <p/>
 * 实现功能：向kafka消息服务发送消息的生产者
 * <p>
 * ------------------------------------------------------------------------------------------------
 * 版本          变更时间             变更人                     变更原因
 * ------------------------------------------------------------------------------------------------
 * 1.0.00      2017-04-12 17:06     秦有宝               新建
 * <p/>
 * *************************************************************************************************
 */
public class MqProducer<M> {

    /**
     * 消息的类型(生产和消费保持一致)
     */
    private String topic;
    //kafka参数配置
    private static final String MQ_CONFIG_KEY="MQ_PRODUCER_CONFIG";
    private static Properties kafkaProperties;
    /**
     * 构造函数
     * @param topic 消息的类型
     */
    public MqProducer(String topic){
        this.topic = topic;
        //获取kafka全局参数配置
//        kafkaProperties = GlobalParam.getProperties(MQ_CONFIG_KEY);
//        if (kafkaProperties==null){
//            throw new ExceptionInInitializerError("系统没有配置kafka的生产者全局参数："+MQ_CONFIG_KEY);
//        }


    }

    private KafkaProducer<String,String> createProducer(){
        return new KafkaProducer<>(kafkaProperties);
    }

    /**
     * 将消息对象序列化为JSON字符串
     * @param message 消息
     * @return JSON字符串
     */
    private String toJson(M message){
        try {
            return JsonUtils.toJson(message);
        } catch (Exception e) {
            e.printStackTrace();
        }
        return null;
    }

    /**
     * 发送一个消息(同步发送)
     * @param key 消息的键值
     * @param message 消息
     */
    public void send(String key,M message){
        KafkaProducer<String,String> producer = new KafkaProducer<>(kafkaProperties);
        ObjectMapper objectMapper = new ObjectMapper();
        String value = null;
        try {
            value = objectMapper.writeValueAsString(message);
        } catch (JsonProcessingException e) {
            e.printStackTrace();
        }
        if (value==null)
        {
            return;
        }
        try {
            producer.send(new ProducerRecord<>(topic, key, value)).get();
        } catch (InterruptedException | ExecutionException e) {
            e.printStackTrace();
        }
        producer.close();
    }

    /**
     * 发送一个消息(异步发送)
     * @param key 消息的键值
     * @param message 消息
     */
    public void sendAsyn(String key,M message) {
        KafkaProducer<String,String> producer = createProducer();
        producer.send(new ProducerRecord<>(topic, key, toJson(message)));
        producer.close();
    }

    /**
     * 发送一批消息（同步发送）
     * @param messages 发送的消息
     */
    public void send(Map<String,M> messages){
        if (messages.isEmpty()){
            return;
        }
        //循环发送
        KafkaProducer<String,String> producer = createProducer();
        messages.forEach((k,m)->{
            ProducerRecord<String,String> record = new ProducerRecord<>(topic,k,toJson(m));
            try {
                producer.send(record).get();
            } catch (InterruptedException | ExecutionException e) {
                e.printStackTrace();
            }
        });
        producer.close();
    }

    /**
     * 发送一批消息（异步发送）
     * @param messages 发送的消息
     */
    public void sendAsyn(Map<String,M> messages) {
        if (messages.isEmpty()) {
            return;
        }
        //循环发送
        KafkaProducer<String, String> producer = createProducer();
        messages.forEach((k, m) -> {
            ProducerRecord<String, String> record = new ProducerRecord<>(topic, k, toJson(m));
            producer.send(record);
        });
        producer.close();
    }
}
